package com.bigdata.core.transformations

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
 * aggregateByKey
 * 首先是给定RDD的每个分区一个初始值，然后RDD中每个分区中按照相同的key，结合初始值去合并，最后RDD之间相同的key 聚合。
 */
object Demo26_AggregateByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("aggregateByKey").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val rdd1: RDD[(String, Int)] = sc.makeRDD(List[(String, Int)](
      ("zhangsan", 10), ("zhangsan", 20), ("wangwu", 30),
      ("lisi", 40), ("zhangsan", 50), ("lisi", 60),
      ("wangwu", 70), ("wangwu", 80), ("lisi", 90)
    ), 3)
    // 查看每个分区的数据
    val count: Long = rdd1.mapPartitionsWithIndex((index, iter) => {
      val arr: ArrayBuffer[(String, Int)] = ArrayBuffer[(String, Int)]()
      iter.foreach(elem => {
        println("rdd1 partition index = " + index + ", value = " + elem)
        arr.append(elem)
      })
      arr.iterator
    }).count()
    println("count： " + count)

    /**
     * 0号分区：
     * ("zhangsan",10) ("zhangsan",20) ("wangwu",30)
     * 1号分区：
     * ("lisi",40) ("zhangsan",50) ("lisi",60)
     * 2号分区：
     * ("wangwu",70) ("wangwu",80) ("lisi",90)
     *
     * 经过第二个参数，map端聚合 :
     * 0:("zhangsan",hello~10~20),("wangwu",hello~30)
     * 1:("zhangsan",hello~50),("lisi"，hello~40~60)
     * 2:("lisi",hello~90)，("wangwu",hello~70~80)
     *
     * 分区合并后：("zhangsan",hello~10~20#hello~50),
     * ("lisi",hello~40~60#hello~90),
     * ("wangwu",hello~30#hello~70~80),
     *
     * "hello" 是初始值, 也是后面函数中的acc
     * 第一个函数是用来在每个分区内对键值对进行聚合的函数, v是RDD中传递的值
     * 第二个函数  是用来在不同分区之间合并聚合结果的函数。
     */
    val result: RDD[(String, String)] = rdd1.aggregateByKey("hello")((acc, v) => {
      acc + "~" + v
    }, (acc1, acc2) => {
      acc1 + "#" + acc2
    })
    result.foreach(println)

  }
}
